package com.fary.buffer;

import cn.hutool.json.JSONUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

import static com.fary.constant.RabbitMQExchangeConstant.WECHAT_BATCH_TEMPLATE_EXCHANGE;

@Slf4j
@Component
public class MqBatchDeliveryBuffer<E> {
    /**
     * 定义容器存放msg
     */
    private final LinkedBlockingQueue<E> buffers = new LinkedBlockingQueue<E>();
    private boolean isRun = true;

    @Autowired
    private RabbitTemplate rabbitTemplate;

    public MqBatchDeliveryBuffer() {
        // 利用单例线程池启动该线程
        Executors.newSingleThreadExecutor().execute(new TaskBufferThread());
    }

    /**
     * 将msg消息暂时存放在我们缓冲区中
     */
    public void add(E e) {
        buffers.offer(e);
    }

    /**
     * 合并消息线程（将多条消息合并成一条msg ，投递MQ中）
     */
    class TaskBufferThread implements Runnable {
        @Override
        public void run() {
            while (isRun) {
                try {
                    // 合并多少条消息
                    List<E> tempBuffer = new ArrayList<>();
                    for (int i = 0; i < 5; i++) {
                        // 超时时间是为1s 1s 没有取到情况下 则 返回null
                        E e = buffers.poll(1000L, TimeUnit.SECONDS);
                        if (e != null) {
                            tempBuffer.add(e);
                        }
                    }
                    // 批量的形式将数据投递到MQ中
                    String msg = JSONUtil.toJsonStr(tempBuffer);
                    log.info("[生产者批量的方式获取数据，投递到MQ中,sendMsgs:{}]", msg);
                    rabbitTemplate.convertAndSend(WECHAT_BATCH_TEMPLATE_EXCHANGE, "", msg);
                } catch (Exception e) {
                    log.error("e:{}", e);
                }
            }
        }
    }


}
